var db = require("./libs/db"); 
var mqtt = require('mqtt');
var path = require('path');
var fs = require('fs');
var uuidv4 = require('uuid');
var format = require('string-format');
format.extend(String.prototype);

const kurento = require("kurento-client"); 
const { Session } = require("inspector");

class THHMqKmsChannel{
    constructor(){
        this.chId="";
        this.pipeId="";
        this.candidateQueue ={};
        this.sessions={};
        this.players={};
    }
}
class THhMqKmsV10{
    constructor(){
        this.config = {
            appId:"THhMqKmsV10",
            dataPath:"D:/gitee/vsport/vsport/svr/config/hh-mq-kms.txt" ,
            kmsUrl:"wss://wss.hhdata.cn:8433/kurento",
            mq:{
                url:"tcp://gmggf8g.mqtt.iot.bj.baidubce.com:1883",
                uid:"gmggf8g/peer",
                pwd:"a52bwKg22dCAuQzB" 
            }
        }
        this.channels={};
        this.objs={}; 
        this.kmsClient = null ;
        this.kmsClientTimer = null ;
        this.mqClient = null;
        this.isSaving = false;
    }
    log( logData ){
        console.log(logData);
    }
    open(){
        var me = this;
        return new Promise((success , faild)=>{
            try{
                me.openMq().then(()=>{
                    return me.getKmsClient();
                }).then( kmsClient=>{
                    return me.loadKmsObjs(kmsClient);
                }).then(()=>{
                    success();
                });
            }
            catch(er){
                me.log(er);
                success();
            }
        })
    }
    openMq(){
        var me = this;
        return new Promise((success , faild)=>{
            try{
                var options={
                    username: me.config.mq.uid ,
                    password: me.config.mq.pwd ,
                    clientId: me.config.clientId
                };
                var client=mqtt.connect( me.config.mq.url , options );
                me.mqClient = client;
                                    
                client.on('connect', function () {
                    console.log("connected to broker") ;
                })
                client.on('close',function(){
                    console.log("connection closed");
                })
                client.on('message', function (topic, message) {
                    var msg=JSON.parse(message.toString());    
                    try{
                        if( msg.msgId=="offline"){
                            ;
                        }
                        else if( msg.msgId=="acLoadChannel"){ 
                            me.acLoadChannel( msg);
                        }
                        else if( msg.msgId=="acCreateSession"){ 
                            me.acCreateSession( msg);
                        }
                        else if( msg.msgId=="acCreateSessionEnd"){ 
                            me.acCreateSession( msg);
                        }
                        else if( msg.msgId=="acLoadPlayer"){ 
                            me.acLoadPlayer( msg); 
                        }
                        else if( msg.msgId=="acConnectMedia"){
                            me.acConnectMedia( msg);  
                        }
                        else {
                            console.log(msg)
                        }
                    }
                    catch(exr){
                        console.log(exr)
                    }
                }); 
                var topic = 'hh/'+me.config.appId+"/svr/#";
                client.subscribe([topic]); 
                console.log('mq topic :' , topic);
                success();
            }
            catch(er){
                me.log(er);
                success();
            }
        })
    }
    getKmsClient(){
        var me = this;
        return new Promise((success , faild)=>{
            try{
                var freeClient = function(client){
                    try{
                        client = null;
                        clearInterval(me.kmsClientTimer);
                    }
                    catch(err){
                        me.log(eer); 
                    }
                }
                var handler = function(client){
                    try{
                        if( client && client!=null){
                            client.getServerManager( function(erSvr , svr){
                                if( svr==null || !svr){
                                    freeClient(client);
                                }
                            })
                        }
                        else{
                            freeClient(client);
                        }
                    }
                    catch(err){
                        me.log(eer);
                        freeClient(client);
                    }
                }
                var createClient = function(){
                    kurento( me.config.kmsUrl, function(erClienrt ,client){
                        me.kmsClient = client;
                        me.kmsClientTimer = setInterval( handler , 60*1000 , client);
                        success(client);
                    })
                }
                if( me.kmsClient && me.kmsClient!=null){
                    success( me.kmsClient);
                }
                else{
                    createClient();
                }
            }
            catch(er){
                me.log(er);
                success();
            }
        })
    }
    loadKmsObjs(kmsClient){
        var me = this;
        return new Promise((success , faild)=>{
            try{
                me.objs ={};
                if( fs.existsSync(me.config.dataPath)){
                    var content = fs.readFileSync(me.config.dataPath);
                    var infos = JSON.parse(content);
                    var loadItem = function(){
                        if( infos.length>0){
                            var info = infos.shift();
                            kmsClient.getMediaobjectById( info.kmsId, function(erFind , obj){
                                if( obj && obj!=null){
                                    obj.info = info;
                                    me.objs[obj.id] = obj;
                                }
                                loadItem();
                            })
                        }
                        else{
                            me.saveKmsObjs();
                            success();
                        }
                    }
                    loadItem();
                }
            }
            catch(er){
                me.log(er);
                success();
            }
        })
    }
    saveKmsObjs( ){
        var me = this;
        try{
            while( !me.isSaving){
                me.isSaving  = true;
                try{
                    var infos=[];
                    Object.values(me.objs).forEach(obj=>{
                        infos.push(obj.info);
                    })
                    var content = JSON.stringify(infos);
                    fs.writeFileSync( me.config.dataPath , content);
                }
                catch(eee){
                    me.log(eee);
                }
                me.isSaving = false;
            }
        }
        catch(er){
            me.log(er);
            success();
        }
    }
    registObj( obj , channelId , clientId , code  ){
        var me = this;
        return new Promise((success , faild)=>{
            try{
                if( obj){
                    var info = {
                        channelId: channelId ,
                        clientId : clientId ,
                        code: code ,
                        kmsId: obj.id
                    };
                    obj.info = info;
                    me.objs[obj.id] = obj;
                    me.saveKmsObjs();
                }
            }
            catch(er){
                me.log(er);
                success();
            }
        })
    } 
    responseMsg( clientId , reqId , msgId , msgBody){
        var me = this;
        try{
            var topic = 'hh/'+me.config.appId+"/"+ clientId;
            var msg ={
                id: msgId ,
                reqId: reqId ,
                clientId: clientId ,
                body:msgBody
            }
            me.mqClient.publish(topic , JSON.stringify(msg));
        }
        catch(er){
            me.log(er);
            success();
        }
    }
    notifyMsg( clientId , msgId , msgBody){
        var me = this;
        try{
            var topic = 'hh/'+me.config.appId+"/"+ clientId;
            var msg ={
                id: msgId ,
                clientId: clientId ,
                body:msgBody
            }
            me.mqClient.publish(topic , JSON.stringify(msg));
        }
        catch(er){
            me.log(er);
            success();
        }
    }
    acLoadChannel( msg ){
        var me = this;
        try{
            var chId = msg.body.chId;
            var channel = me.channels[chId];
            if(!channel){
                channel = new THHMqKmsChannel();
                channel.chId = chId;
                me.channels[chId] = channel;
            }
            me.getKmsClient().then(client=>{
                client.create("MediaPipeline" , function(erPipe , pipe){
                    me.registObj(pipe , channelId , 'svr',  'ch-pipe-'+ channelId ).then(pipeObj=>{
                        channel.pipeId = pipe.id;
                        me.responseMsg( msg.clientId , msg.reqId , "respLoadChannel" , channel);
                    })
                })
            })
        }
        catch(er){
            me.log(er); 
        }
    }
    acCreateSession( msg ){
        var me = this;
        try{
            var msgBody={
                state :0 ,
                session: null
            }
            var chId = msg.body.chId;
            var clientId = msg.clientId;
            var endCode = msg.body.endCode;
            var reqId = msg.reqId; 
            var ch = me.channels[chId];
            if( ch){
                var pipe = me.objs[ch.pipeId];
                pipe.create("WebRtcEndpoint", function(erEnd , end){
                    me.registObj(end , chId , clientId , enCode).then(endObj=>{
                        end.on('OnIceCandidate', function(event) {
                            var candidate = kurento.getComplexType('IceCandidate')(event.candidate);
                            var msgBody = {
                                sessionId: session.sessionId ,
                                candidate: candidate,
                                endId: end.id
                            }
                            me.notifyMsg( clientId , reqId , "endCandidate" , msgBody);
                        });
                        var session = {
                            sessionId: uuidv4(),
                            chId:chId , 
                            clientId: clientId ,
                            endCode:endCode ,
                            endId: end.id
                        };
                        ch.sessions[sessionId] = session;
                        msgBody.session = session;
                        msgBody.state = 1;
                        me.responseMsg( msg.clientId , msg.reqId , "respCreateSession" , msgBody);

                    })
                })
            }
            else{
                msgBody.error ="chId is error.";
                me.responseMsg( msg.clientId , msg.reqId , "respCreateSession" , msgBody);
            }
        }
        catch(er){
            me.log(er);
            success();
        }
    }
    acEndOffer( msg ){
        var me = this;
        return new Promise((success , faild)=>{
            try{
                var msgBody={
                    state :0 ,
                    session: null
                }
                var chId = msg.body.chId;
                var clientId = msg.clientId;
                var reqId = msg.reqId;
                var sessionId = msg.body.sessionId;
                var offer = msg.body.sdpOffer;
                var ch = me.channels[chId];
                if( ch){
                    var session = ch.success[sessionId];
                    if( session) {
                        var end = me.objs[session.endId];
                        end.processOffer( offer, function(erAnswer , sdpAnswer){  
                            end.gatherCandidates(function(ere){
                                msgBody.session = sesssion;
                                msgBody.sdpAnswer = sdpAnswer;
                                me.responseMsg( clientId , reqId , "respEndOffer" , msgBody);
                            })
                        })
                    }
                    else{
                        msgBody.error ="sessionId is error.";
                        me.responseMsg( clientId , reqId , "respEndOffer" , msgBody);
                    }
                }
                else{
                    msgBody.error ="chId is error.";
                    me.responseMsg( msg.clientId , msg.reqId , "respEndOffer" , msgBody);
                }
            }
            catch(er){
                me.log(er);
                success();
            }
        })
    }
    acLoadPlayer( msg ){
        var me = this;
        try{
            var msgBody={
                state :0 ,
                player: null
            }
            var chId = msg.body.chId;
            var clientId = msg.clientId;
            var endCode = msg.endCode;
            var srcUr = msg.url;
            var reqId = msg.reqId; 
            var ch = me.channels[chId];
            if( ch){
                var player = ch.players[endCode];
                if( player && player!=null){
                    msgBody.state = 1;
                    msgBody.player = player;
                    me.responseMsg( clientId , reqId , "respLoadPlayer" , msgBody);
                }
                else{
                    var pipe = me.objs[ch.pipeId];
                    pipe.create("PlayerEndpoint" , {uri: srcUr}, function(erEnd , end){
                        me.registObj(end , chId , clientId , enCode).then(endObj=>{
                            endObj.on("EndOfStream", function(event){
                                endObj.play();
                            });
                            endObj.play();
                            player={
                                chId: chId ,
                                endCode: endObj,
                                endId: end.id
                            }
                            msgBody.player = player;
                            msgBody.state =1;
                            me.responseMsg( clientId , reqId , "respLoadPlayer" , msgBody);
                        })
                    })
                }
            }
            else{
                msgBody.error ="chId is error.";
                me.responseMsg( clientId , reqId , "respLoadPlayer" , msgBody);
            }
        }
        catch(er){
            me.log(er);
            success();
        }
    }
    acConnectMedia( msg ){
        var me = this;
        try{                
            var srcId = msg.body.srcId;
            var tarId = msg.body.targetId;
            var mediaType = msg.body.mediaType;
            mediaType =(mediaType=="AUDIO"||mediaType=="VIDEO")?mediaType:"";
            var src = me.objs[srcId];
            var target = me.objs[tarId];
            if(src && src!=null && target && target!=null){
                if( mediaType==""){
                    src.connect(target, function(erC){
                        me.log(erC);
                    })
                }
                else{
                    src.connect(target, mediaType, function(erC){
                        me.log(erC);
                    })
                }
            }
        }
        catch(er){
            me.log(er);
            success();
        }
    }
}